package com.road.yishi.execaction;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Description :执行action队列的线程池, 延迟执行的action，先放置到delay action队列中，延迟时间后再加入执行队列
 * @FileName: Executor.java
 * @Author :WeiHui.Zhang
 * @Data : 2016年2月1日 下午2:17:16
 * @Version:V1.00
 */
public final class Executor {

	private final static Logger LOG = LoggerFactory.getLogger(Executor.class);

	private AbstractActionQueue defaultQueue;
	private ThreadPoolExecutor pool;
	private DelayCheckThread delayCheckThread;

	/**
	 * 
	 * @param corePoolSize池中所保存的线程数，包括空闲线程。
	 * @param maximumPoolSize
	 *            池中允许的最大线程数。
	 * @param keepAliveTime当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。
	 * @param cacheSize
	 *            执行队列大小
	 * @param prefix
	 *            线程池前缀名称
	 * @param unit
	 *            参数的时间单位。
	 * @param workQueue
	 *            执行前用于保持任务的队列。此队列仅由保持 execute 方法提交的 Runnable 任务。
	 */
	public Executor(int corePoolSize, int maximumPoolSize, long keepAliveTime, int cacheSize, String prefix,
			TimeUnit unit, BlockingQueue<Runnable> workQueue) {

		// 可选配置
		// 线程名称前缀
		if (prefix == null) {
			prefix = "";
		}
		ThreadFactory threadFactory = new CustomThreadFactory(prefix);
		RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();

		pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue,
				threadFactory, handler);

		defaultQueue = new AbstractActionQueue(this);

		delayCheckThread = new DelayCheckThread(prefix);
		delayCheckThread.start();
	}

	/**
	 * 
	 * @param corePoolSize
	 *            最小线程数，包括空闲线程。
	 * @param maximumPoolSize
	 *            最大线程数
	 * @param keepAliveTime
	 *            当线程数大于最小时，终止多余的空闲线程等待新任务的最长时间
	 * @param cacheSize
	 *            执行队列大小
	 * @param prefix
	 *            线程池前缀名称
	 * @param handler
	 *            由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序。
	 */
	public Executor(int corePoolSize, int maximumPoolSize, long keepAliveTime, int cacheSize, String prefix,
			RejectedExecutionHandler handler) {
		System.out.println("第一步");
		BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();
		// 可选配置
		// 线程名称前缀
		if (prefix == null) {
			prefix = "";
		}
		ThreadFactory threadFactory = new CustomThreadFactory(prefix);
		// 超出线程范围和队列容量而使执行被阻塞时所使用的处理程序
		if (handler == null) {
			handler = new ThreadPoolExecutor.DiscardPolicy();
		}

		pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue,
				threadFactory, handler);

		defaultQueue = new AbstractActionQueue(this);

		delayCheckThread = new DelayCheckThread(prefix);
		delayCheckThread.start();
	}

	public AbstractActionQueue getDefaultQueue() {
		return defaultQueue;
	}

	public void enDefaultQueue(Action action) {
		this.defaultQueue.enqueue(action);
	}

	public void execute(Action action) {
		pool.execute(action);
	}

	public void enDelayQueue(DelayAction delayAction) {
		delayCheckThread.addAction(delayAction);
	}

	public void stop() {
		delayCheckThread.stopping();
		if (!pool.isShutdown()) {
			pool.shutdown();
		}
	}

	static final class CustomThreadFactory implements ThreadFactory {

		private static final AtomicInteger poolNumber = new AtomicInteger(1);
		private final ThreadGroup group;
		private final AtomicInteger threadNumber = new AtomicInteger(1);
		private final String namePrefix;

		CustomThreadFactory(String prefix) {
			SecurityManager securitymanager = System.getSecurityManager();
			group = securitymanager == null ? Thread.currentThread().getThreadGroup()
					: securitymanager.getThreadGroup();
			namePrefix = (new StringBuilder()).append("pool-").append(poolNumber.getAndIncrement()).append("-")
					.append(prefix).append("-thread-").toString();
		}

		@Override
		public Thread newThread(Runnable runnable) {
			Thread thread = new Thread(group, runnable,
					(new StringBuilder()).append(namePrefix).append(threadNumber.getAndIncrement()).toString(), 0L);
			if (thread.isDaemon())
				thread.setDaemon(false);
			if (thread.getPriority() != Thread.NORM_PRIORITY)
				thread.setPriority(Thread.NORM_PRIORITY);
			return thread;
		}

	}

	class DelayCheckThread extends Thread {

		private static final int FRAME_PER_SECOND = 120;
		private Object lock; // 线程锁
		private List<DelayAction> queue;
		private List<DelayAction> execQueue;
		private boolean isRunning;

		public DelayCheckThread(String prefix) {
			super(prefix + "-DelayCheckThread");
			lock = this;
			queue = new ArrayList<DelayAction>();
			execQueue = new ArrayList<DelayAction>();
			isRunning = true;
			setPriority(Thread.MAX_PRIORITY); // 给予高优先级
		}

		public boolean isRunning() {
			return isRunning;
		}

		public void stopping() {
			if (isRunning) {
				isRunning = false;
			}
		}

		@Override
		public void run() {
			System.out.println("第二步");
			long balance = 0;
			while (isRunning) {
				try {
					int execute = 0;
					// 读取待执行的队列,如果没有可以执行的动作则等待
					poll();
					if (execQueue.size() == 0) {
						continue;
					}

					long start = System.currentTimeMillis();
					execute = execActions();
					execQueue.clear();
					long end = System.currentTimeMillis();
					long interval = end - start;
					balance += FRAME_PER_SECOND - interval;
					if (interval > FRAME_PER_SECOND) {
						LOG.warn("DelayCheckThread is spent too much time: " + interval + "ms, execute = " + execute);
					}
					if (balance > 0) {
						Thread.sleep((int) balance);
						balance = 0;
					} else {
						if (balance < -1000) {
							balance += 1000;
						}
					}
				} catch (Exception e) {
					LOG.error("DelayCheckThread error. ", e);
				}
			}
		}

		/**
		 * 返回执行成功的Action数量
		 **/
		public int execActions() {
			int executeCount = 0;
			for (DelayAction delayAction : execQueue) {
				try {
					long begin = System.currentTimeMillis();
					if (delayAction == null) {
						LOG.error("error");
					}
					if (!delayAction.tryExec(begin)) {
						addAction(delayAction);
					}
					executeCount++;
					long end = System.currentTimeMillis();
					if (end - begin > FRAME_PER_SECOND) {
						LOG.warn(delayAction.toString() + " spent too much time. time :" + (end - begin));
					}
				} catch (Exception e) {
					LOG.error("执行action异常" + delayAction.toString(), e);
				}
			}
			return executeCount;
		}

		/**
		 * 添加Action到队列
		 * 
		 * @param delayAction
		 */
		public void addAction(DelayAction delayAction) {
			System.out.println("第5步");
			synchronized (lock) {
				queue.add(delayAction);
				lock.notifyAll();
			}
		}

		/**
		 * 以阻塞的方式读取队列,如果队列为空则阻塞
		 * 
		 * @throws InterruptedException
		 **/
		private void poll() throws InterruptedException {
			System.out.println("第3步");
			synchronized (lock) {
				if (queue.size() == 0) {
					System.out.println("before lock wait");
					lock.wait();
					System.out.println("第6步");
					System.out.println("after lock wait");
				} else {
					System.out.println("第6步");
					execQueue.addAll(queue);
					queue.clear();
					lock.notifyAll();
				}
			}
		}
	}
}
